package com.lj.queue.utils;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import com.lj.queue.entity.DelayQueueMessage;
import com.lj.queue.entity.DelayQueueMeta;
import com.lj.queue.enums.DelayQueueMessageStatus;
import com.lj.queue.handler.*;
import com.lj.queue.service.DelayQueueMessageService;
import com.lj.queue.service.DelayQueueMetaService;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;

/**
 * @author luojing
 * @since 2025/7/17 17:47
 * 延迟队列工具类
 */
@Slf4j
public class DelayQueueUtils {


    private static final Map<String, DelayQueue<DelayedMessage>> delayQueueCache = new ConcurrentHashMap<>();

    private static final ExecutorService delayQueueWorker = SpringUtil.getBean("delayQueueWorker");

    private static final DelayQueueMetaService queueMetaService = SpringUtil.getBean(DelayQueueMetaService.class);
    private static final DelayQueueMessageService queueMessageService = SpringUtil.getBean(DelayQueueMessageService.class);

    public static DelayQueue<DelayedMessage> createDelayQueue(DelayQueueMeta delayQueueMeta, Collection<DelayMessageHandler<Object>> handlerList) {
        String queueName = delayQueueMeta.getName();
        DelayQueue<DelayedMessage> delayQueue = delayQueueCache.get(queueName);
        if (delayQueue != null) {
            return delayQueue;
        }
        DelayQueue<DelayedMessage> newDelayQueue = new DelayQueue<>();
        delayQueueCache.put(queueName, newDelayQueue);
        // 启动时预加载一次
        DelayQueueUtils.pushQueue(newDelayQueue, DelayedMessage.of(LocalDateTime.now())
                .setQueueId(delayQueueMeta.getId())
                .setMessage(JSONUtil.toJsonStr(new PreloadHandler.PreloadMessage(true, delayQueueMeta.getVersion()))));
        // 单独拿一个线程来阻塞监听
        ThreadUtil.execute(() -> {
            try {
                while (true) {
                    DelayedMessage delayedMessage = newDelayQueue.take();
                    // 判断队列是否还在，有没有被删除
                    if (delayQueueCache.containsKey(queueName)) {
                        // 异步执行
                        asyncWork(queueName, handlerList, delayedMessage, newDelayQueue);
                    } else {
                        // 当前队列被删除了，清空队列剩余的消息，然后当前线程就可以结束了
                        log.info("队列({})已被移除", queueName);
                        newDelayQueue.clear();
                        break;
                    }

                }
            } catch (InterruptedException e) {
                log.warn("队列:{} 执行被中断", queueName, e);
            }
        });
        return newDelayQueue;
    }

    private static void asyncWork(String queueName, Collection<DelayMessageHandler<Object>> handlerList, DelayedMessage delayedMessage, DelayQueue<DelayedMessage> delayQueue) {
        delayQueueWorker.execute(() -> {
            // 更新消息状态
            if (delayedMessage.getMessageId() != null) {
                // 预加载的消息是没有id的因为不会在数据库中生成
                // 这里修改状态的时候判断了原来的状态，在分布式部署下防止重复消费的问题
                boolean updated = queueMessageService.updateStatus(delayedMessage.getMessageId(), DelayQueueMessageStatus.trigger);
                if (!updated) {
                    return;
                }
            }
            // 调用执行器责任链
            new DelayHandlerChain(handlerList).next(DelayHandlerContext.of(queueName, delayedMessage, delayQueue,
                    queueMetaService, queueMessageService));
        });
    }

    public static DelayQueue<DelayedMessage> getQueue(String queueName) {
        return delayQueueCache.get(queueName);
    }

    // =======================pushQueue=======================
    // 操作会直接操作延迟队列,存在消息重启丢失的风险，非必要请使用pushMessage方法

    public static void pushQueue(String queueName, DelayQueueMessage message) {
        pushQueue(getQueue(queueName), message);
    }

    public static void pushQueue(DelayQueue<DelayedMessage> delayQueue, DelayQueueMessage message) {
        pushQueue(delayQueue, DelayedMessage.of(message));
    }

    public static void pushQueue(DelayQueue<DelayedMessage> delayQueue, DelayedMessage delayedMessage) {
        if (delayQueue != null) {
            delayQueue.add(delayedMessage);
        }
    }

    public static void pushQueue(String queueName, Collection<DelayQueueMessage> messageList) {
        pushQueue(getQueue(queueName), messageList);
    }

    public static void pushQueue(DelayQueue<DelayedMessage> delayQueue, Collection<DelayQueueMessage> messageList) {
        if (delayQueue == null || CollUtil.isEmpty(messageList)) {
            return;
        }
        delayQueue.addAll(messageList.stream().filter(Objects::nonNull).map(DelayedMessage::of).toList());
    }
    // =======================pushQueue=======================

    // =======================pushMessage======================

    public static boolean pushMessage(String queueName, DelayQueueMessage message) {
        return pushMessage(queueName, Collections.singletonList(message));
    }

    public static boolean pushMessage(String queueName, List<DelayQueueMessage> messageList) {
        return pushMessage(queueMetaService.getByQueueName(queueName), messageList);
    }

    public static boolean pushMessage(DelayQueueMeta queueMeta, DelayQueueMessage message) {
        return pushMessage(queueMeta, Collections.singletonList(message));
    }

    public static boolean pushMessage(DelayQueueMeta queueMeta, List<DelayQueueMessage> messageList) {
        return queueMessageService.pushMessage(queueMeta, messageList);
    }

    // =======================pushMessage======================

    public static void removeQueue(String queueName) {
        delayQueueCache.remove(queueName);
    }
}
